package sync

import (
	"context"
	"database/sql"
	"datasync/config"
	"datasync/db"
	"fmt"
	"log"
	"strconv"
	"sync"
	"time"

	"github.com/go-redis/redis/v8"
)

// SyncService 同步服务
type SyncService struct {
	SourceDB      *sql.DB
	TargetDB      *sql.DB
	Config        *config.SyncConfig
	RedisClient   *redis.Client          // Redis客户端
	lastSyncTimes map[string]interface{} // 记录每个表的最后同步主键值
	mu            sync.Mutex             // 保护lastSyncTimes的互斥锁
	ctx           context.Context        // Redis上下文
}

// NewSyncService 创建同步服务实例
func NewSyncService(sourceDB, targetDB *sql.DB, cfg *config.SyncConfig) *SyncService {
	ctx := context.Background()
	var redisClient *redis.Client

	// 如果配置了Redis，则初始化Redis客户端
	if cfg.Redis != nil {
		redisClient = redis.NewClient(&redis.Options{
			Addr:     fmt.Sprintf("%s:%d", cfg.Redis.Host, cfg.Redis.Port),
			Password: cfg.Redis.Password,
			DB:       cfg.Redis.DB,
		})

		// 测试Redis连接
		_, err := redisClient.Ping(ctx).Result()
		if err != nil {
			log.Printf("Redis连接失败: %v, 将使用内存存储同步状态", err)
			redisClient = nil
		} else {
			log.Println("Redis连接成功，将使用Redis存储同步状态")
		}
	}

	return &SyncService{
		SourceDB:      sourceDB,
		TargetDB:      targetDB,
		Config:        cfg,
		RedisClient:   redisClient,
		ctx:           ctx,
		lastSyncTimes: make(map[string]interface{}),
	}
}

// Start 启动同步服务
func (s *SyncService) Start() error {
	log.Println("启动数据同步服务...")

	// 验证所有表映射的表结构
	for _, mapping := range s.Config.TableMappings {
		if !mapping.SyncEnabled {
			log.Printf("表映射 %s->%s 已禁用，跳过验证", mapping.SourceTable, mapping.TargetTable)
			continue
		}

		if err := s.validateTableSchema(mapping.SourceTable, mapping.TargetTable); err != nil {
			log.Printf("表结构验证失败 %s->%s: %v", mapping.SourceTable, mapping.TargetTable, err)
			return err
		}

		// 启动定时同步协程
		go s.startTableSync(mapping)
	}

	return nil
}

// validateTableSchema 验证源表和目标表结构兼容性
func (s *SyncService) validateTableSchema(sourceTable, targetTable string) error {
	// 检查源表是否存在
	exists, err := db.TableExists(s.SourceDB, sourceTable)
	if err != nil {
		return fmt.Errorf("检查源表是否存在失败: %v", err)
	}
	if !exists {
		return fmt.Errorf("源表 %s 不存在", sourceTable)
	}

	// 获取源表结构
	sourceSchema, err := db.GetTableSchema(s.SourceDB, sourceTable)
	if err != nil {
		return fmt.Errorf("获取源表 %s 结构失败: %v", sourceTable, err)
	}
	_ = sourceSchema // 标记变量已使用，避免编译错误

	// 检查目标表是否存在，如果不存在则创建
	exists, err = db.TableExists(s.TargetDB, targetTable)
	if err != nil {
		return fmt.Errorf("检查目标表是否存在失败: %v", err)
	}

	if !exists {
		log.Printf("目标表 %s 不存在，正在从源表 %s 创建...", targetTable, sourceTable)
		if err := db.CreateTableFromSource(s.SourceDB, s.TargetDB, sourceTable, targetTable); err != nil {
			return fmt.Errorf("创建目标表失败: %v", err)
		}
	} else {
		// 如果目标表存在，同步表结构
		if err := db.SyncTableSchema(s.SourceDB, s.TargetDB, sourceTable, targetTable); err != nil {
			return fmt.Errorf("同步表结构失败: %v", err)
		}
	}

	log.Printf("表结构验证通过: %s->%s", sourceTable, targetTable)
	return nil
}

// startTableSync 启动表同步任务
func (s *SyncService) startTableSync(mapping config.TableMapping) {
	log.Printf("启动表同步任务: %s->%s, 间隔 %d 秒", mapping.SourceTable, mapping.TargetTable, mapping.SyncInterval)

	// 立即执行一次同步
	s.syncTableData(mapping)

	// 设置定时任务
	ticker := time.NewTicker(time.Duration(mapping.SyncInterval) * time.Second)
	defer ticker.Stop()

	for range ticker.C {
		s.syncTableData(mapping)
	}
}

// syncTableData 同步单表数据
func (s *SyncService) syncTableData(mapping config.TableMapping) {
	startTime := time.Now()
	log.Printf("开始同步表数据: %s->%s", mapping.SourceTable, mapping.TargetTable)

	// 获取增量同步参数
	var lastSyncPrimaryKey interface{}
	if mapping.IncrementalSync {
		// 优先从Redis获取上次同步主键
		if s.RedisClient != nil && mapping.PrimaryKeyField != "" {
			redisKey := fmt.Sprintf("sync:last_sync:%s", mapping.SourceTable)
			val, err := s.RedisClient.Get(s.ctx, redisKey).Result()
			if err == nil && val != "" {
				// 转换Redis存储的主键值
			lastSyncPrimaryKey = convertRedisValueToPrimaryKey(val, mapping.SourceTable)
				log.Printf("从Redis获取表 %s 上次同步主键值: %v", mapping.SourceTable, lastSyncPrimaryKey)
			} else if err != redis.Nil {
				log.Printf("Redis获取失败: %v，将使用内存存储值", err)
			}
		}

		// 如果Redis未获取到值，则使用内存存储
		if lastSyncPrimaryKey == nil {
			s.mu.Lock()
			lastSyncPrimaryKey = s.lastSyncTimes[mapping.SourceTable]
			s.mu.Unlock()
		}

		log.Printf("表 %s 启用增量同步，上次同步主键值: %v", mapping.SourceTable, lastSyncPrimaryKey)
	}

	// 获取源表数据
	data, err := s.fetchSourceData(mapping.SourceTable, mapping.IncrementalSync, mapping.PrimaryKeyField, lastSyncPrimaryKey)
	if err != nil {
		log.Printf("获取源表 %s 数据失败: %v", mapping.SourceTable, err)
		return
	}

	if len(data) == 0 {
		log.Printf("源表 %s 没有需要同步的数据", mapping.SourceTable)
		return
	}

	// 写入目标表
	count, err := s.writeTargetData(mapping.TargetTable, data)
	if err != nil {
		log.Printf("写入目标表 %s 数据失败: %v", mapping.TargetTable, err)
		return
	}

	// 更新最后同步主键值
	if mapping.IncrementalSync && mapping.PrimaryKeyField != "" {
		var maxPrimaryKey interface{}
		if len(data) > 0 {
			maxPrimaryKey = data[len(data)-1][mapping.PrimaryKeyField]
		} else if lastSyncPrimaryKey != nil {
			maxPrimaryKey = lastSyncPrimaryKey // 没有新数据，保持上次值
		} else {
			maxPrimaryKey = 0 // 初始值
		}

		// 同时更新内存和Redis
		s.mu.Lock()
		s.lastSyncTimes[mapping.SourceTable] = maxPrimaryKey
		s.mu.Unlock()

		// 如果配置了Redis，同步更新到Redis
		if s.RedisClient != nil {
			redisKey := fmt.Sprintf("sync:last_sync:%s", mapping.SourceTable)
			redisVal := fmt.Sprintf("%v", maxPrimaryKey)
			if err := s.RedisClient.Set(s.ctx, redisKey, redisVal, 0).Err(); err != nil {
				log.Printf("Redis存储失败: %v", err)
			} else {
				log.Printf("已将表 %s 同步状态保存到Redis", mapping.SourceTable)
			}
		}

		log.Printf("更新表 %s 最后同步主键值: %v", mapping.SourceTable, maxPrimaryKey)
	}

	duration := time.Since(startTime)
	log.Printf("表同步完成: %s->%s, 同步记录数: %d, 耗时: %v", mapping.SourceTable, mapping.TargetTable, count, duration)
}

// convertRedisValueToPrimaryKey 将Redis存储的字符串值转换为主键类型
func convertRedisValueToPrimaryKey(val string, tableName string) interface{} {
	// 尝试转换为整数类型
	if intVal, err := strconv.Atoi(val); err == nil {
		return intVal
	}

	// 尝试转换为无符号整数
	if uintVal, err := strconv.ParseUint(val, 10, 64); err == nil {
		return uintVal
	}

	// 尝试转换为浮点数
	if floatVal, err := strconv.ParseFloat(val, 64); err == nil {
		return floatVal
	}

	// 所有转换失败，作为字符串处理并记录警告
	log.Printf("警告: 表 %s 主键值 '%s' 无法转换为数字类型，将作为字符串处理", tableName, val)
	return val
}

// fetchSourceData 获取源表数据
func (s *SyncService) fetchSourceData(tableName string, incrementalSync bool, primaryKeyField string, lastSyncPrimaryKey interface{}) ([]map[string]interface{}, error) {
	var query string
	var rows *sql.Rows
	var err error

	if incrementalSync && primaryKeyField != "" && lastSyncPrimaryKey != nil {
		// 增量同步查询
		query = fmt.Sprintf("SELECT * FROM %s WHERE %s > ?", tableName, primaryKeyField)
		log.Printf("执行增量查询SQL: %s, 上次同步主键值: %v", query, lastSyncPrimaryKey)
		rows, err = s.SourceDB.Query(query, lastSyncPrimaryKey)
	} else {
		// 全量同步查询
		query = fmt.Sprintf("SELECT * FROM %s", tableName)
		log.Printf("执行全量查询SQL: %s", query)
		rows, err = s.SourceDB.Query(query)
	}

	if err != nil {
		return nil, fmt.Errorf("查询源表数据失败: %v", err)
	}
	defer rows.Close()

	// 获取列信息
	columns, err := rows.Columns()
	if err != nil {
		return nil, fmt.Errorf("获取列信息失败: %v", err)
	}

	// 准备扫描目标
	count := len(columns)
	values := make([]interface{}, count)
	valuePtrs := make([]interface{}, count)

	var result []map[string]interface{}

	for rows.Next() {
		// 初始化每一行的指针
		for i := range columns {
			valuePtrs[i] = &values[i]
		}

		// 扫描行数据
		if err := rows.Scan(valuePtrs...); err != nil {
			return nil, fmt.Errorf("扫描行数据失败: %v", err)
		}

		// 转换为map
		row := make(map[string]interface{})
		for i, col := range columns {
			val := values[i]

			// 处理NULL值
			if val == nil {
				row[col] = nil
				continue
			}

			// 处理不同类型
			switch v := val.(type) {
			case []byte:
				row[col] = string(v)
			case int64:
				row[col] = v
			case float64:
				row[col] = v
			case bool:
				row[col] = v
			case time.Time:
				row[col] = v
			default:
				row[col] = fmt.Sprintf("%v", v)
			}
		}

		result = append(result, row)
	}

	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("行迭代错误: %v", err)
	}

	return result, nil
}

// writeTargetData 写入目标表数据
func (s *SyncService) writeTargetData(tableName string, data []map[string]interface{}) (int, error) {
	if len(data) == 0 {
		return 0, nil
	}

	// 开始事务
	tx, err := s.TargetDB.Begin()
	if err != nil {
		return 0, fmt.Errorf("开启事务失败: %v", err)
	}
	defer tx.Rollback()

	// 获取列名
	columns := make([]string, 0, len(data[0]))
	for col := range data[0] {
		columns = append(columns, col)
	}

	// 构建INSERT语句
	placeholders := make([]string, len(columns))
	for i := range placeholders {
		placeholders[i] = "?"
	}

	query := fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s) ON DUPLICATE KEY UPDATE %s",
		tableName,
		joinStrings(columns, ","),
		joinStrings(placeholders, ","),
		buildUpdateClause(columns))
	log.Printf("执行插入SQL: %s", query)

	// 准备语句
	stmt, err := tx.Prepare(query)
	if err != nil {
		return 0, fmt.Errorf("准备语句失败: %v", err)
	}
	defer stmt.Close()

	// 执行批量插入
	count := 0
	for _, row := range data {
		// 准备参数
		params := make([]interface{}, len(columns))
		for i, col := range columns {
			params[i] = row[col]
		}

		// 执行插入
		_, err := stmt.Exec(params...)
		if err != nil {
			return count, fmt.Errorf("执行插入失败: %v, 参数: %v", err, params)
		}

		count++
	}

	// 提交事务
	if err := tx.Commit(); err != nil {
		return count, fmt.Errorf("提交事务失败: %v", err)
	}

	return count, nil
}

// 辅助函数: 连接字符串切片
func joinStrings(strs []string, sep string) string {
	result := ""
	for i, s := range strs {
		if i > 0 {
			result += sep
		}
		result += s
	}
	return result
}

// 辅助函数: 构建UPDATE子句
func buildUpdateClause(columns []string) string {
	clauses := make([]string, len(columns))
	for i, col := range columns {
		clauses[i] = fmt.Sprintf("%s = VALUES(%s)", col, col)
	}
	return joinStrings(clauses, ",")
}
